package org.etnaframework.module.localcache;

import org.etnaframework.module.base.utils.CronExecutorUtils;
import org.etnaframework.module.base.utils.DatetimeUtils;
import org.etnaframework.module.base.utils.NamedThreadFactory;
import org.slf4j.Logger;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.RedisStringCommands;
import org.springframework.data.redis.core.types.Expiration;

import java.nio.charset.StandardCharsets;
import java.util.*;
import java.util.concurrent.*;

/**
 * 全量内存缓存 核心实现
 *
 * <pre>
 * 1、缓存初始化时，完成参数校验赋值和首次加载之后，程序会启动3个定时任务 + 1个监听任务
 *
 *    A.定时全量更新任务，定时向队列{@link #updateTaskQueue}提交任务类型{@link LocalCacheTaskType#SCHEDULED_FULL_UPDATE}
 *    B.检查点更新任务（Type1类型不启用），定时向队列{@link #updateTaskQueue}提交任务类型{@link LocalCacheTaskType#CHECKPOINT_UPDATE}
 *    C.队列消费任务，消费并处理队列{@link #updateTaskQueue}中的任务，并做超时判断处理
 *      当更新任务执行超过配置的时长时，触发{@link ILocalCacheCustomizer#onTimeout(String, long)}
 *
 *        Q: 为什么A和B不直接去调用更新方法，而是通过队列缓冲？
 *        A: 因为这两种更新任务可以合并处理，例如队列中有检查点更新和定时全量更新任务，只需要执行一次定时全量更新任务就可以了
 *           所以C也采用定时调用而不是死循环取的方式，让任务消息有时间积攒
 *
 *    D.监听任务（Type1类型不启用），监听redis的{@link #redisTopic}的消息，收到消息就存到{@link #checkpointMessageQueue}，等待C任务处理
 *      启用队列保存也是为了合并处理，多个增量更新消息和一个全量更新消息可以合并成一个全量更新任务来处理
 *
 * 2、缓存更新事件的处理流程
 *
 *    ①使用者向redis的{@link #redisTopic}发布一条信息msg，内容为需要更新的key，如果想要全量更新则内容为{@link #FULL_UPDATE_KEY}，然后再更新一下{@link #redisKey}的tag
 *
 *        Q: 为什么这里要pub topic后还要set tag？
 *        A: 因为redis的pub/sub的消息没有持久化机制，如果客户端掉线就有可能丢消息，导致缓存未同步更新
 *           针对这种情况的补救措施：消息处理时如果发现key变了，但topic为空，执行一次全量加载
 *           （注意这里也不能100%保证靠谱，因为在断线重连的间隙中也有可能会有增量加载消息覆盖key，所以必须有定时的全量加载兜底）
 *
 *    ②监听任务D，监听{@link #redisTopic}收到消息，保存到{@link #checkpointMessageQueue}
 *      A和B定时投递任务 -> C队列消费任务，将任务合并后进行下一步的处理 {@link #processUpdateTask(boolean)}
 *
 *    ③任务类型为：{@link LocalCacheTaskType#SCHEDULED_FULL_UPDATE}
 *        -> 执行全量加载{@link ILocalCacheCustomizer#load()}
 *           （这里的全量加载不管有没有变化，一旦到点触发就一定会执行，而下面的是有变化才会执行）
 *      任务类型为：{@link LocalCacheTaskType#CHECKPOINT_UPDATE}
 *        -> 检查redis中的{@link #redisKey}是否和本地的一致？
 *             -> 一致，说明没有变化，流程结束
 *             -> 不一致，说明有变化，将本地{@link #checkpointMessageQueue}消息全部取出并做合并处理
 *                  -> 全量加载{@link ILocalCacheCustomizer#load()}
 *                  -> 增量加载{@link ILocalCacheCustomizerType3#onIncrementalKeysReceived(Object, List)}
 * </pre>
 *
 * @author jasyaf
 * @since 2023-11-01
 */
class LocalCacheCore<T> {

    static final Logger log = LocalCache.log;

    static final ScheduledExecutorService dispatcher = Executors.newScheduledThreadPool(Runtime.getRuntime().availableProcessors(), new NamedThreadFactory("LocalCacheDispatchCron::", Thread.MAX_PRIORITY, true));

    static final ExecutorService executor = Executors.newCachedThreadPool(new NamedThreadFactory("LocalCacheCron::", Thread.NORM_PRIORITY, true));

    static final Set<String> CACHE_NAMES = new LinkedHashSet<>();

    static final String CHECKPOINT_UPDATE_KEY_SUFFIX = "-KEY";

    static final String CHECKPOINT_UPDATE_TOPIC_SUFFIX = "-TOPIC";

    static final String FULL_UPDATE_KEY = "";

    static final long REDIS_KEY_EXPIRE_SEC = TimeUnit.DAYS.toSeconds(1);

    static final String PROCESS_UPDATE_TASK_CRON = " */1 * * * * *";

    static final long PROCESS_UPDATE_TASK_DELAY_MS = 500;

    RedisConnectionFactory redisConnectionFactory;

    final String name;

    final String redisKey;

    final String redisTopic;

    final ILocalCacheCustomizer<T> customizer;

    final ILocalCacheCustomizerType3<T> incrementalDataLoader;

    String scheduledFullUpdateCron;

    String checkpointUpdateCron;

    String checkpointUpdateTag;

    final LinkedBlockingQueue<String> checkpointMessageQueue;

    final LinkedBlockingQueue<LocalCacheTaskType> updateTaskQueue;

    final long processUpdateTaskTimeoutMs;

    volatile boolean running;

    volatile long nextTimeoutReportTimeMs;

    volatile long processUpdateTaskStartTimeMs;

    volatile T cache;

    public T get() {
        return cache;
    }

    LocalCacheCore(String name, String scheduledFullUpdateCron, long timeoutSec, ILocalCacheCustomizer<T> customizer, ILocalCacheCustomizerType3<T> incrementalDataLoader) {
        this.name = name;
        this.redisKey = String.format("%s%s", name, CHECKPOINT_UPDATE_KEY_SUFFIX);
        this.redisTopic = String.format("%s%s", name, CHECKPOINT_UPDATE_TOPIC_SUFFIX);
        this.scheduledFullUpdateCron = scheduledFullUpdateCron;
        this.checkpointUpdateTag = "";
        this.checkpointMessageQueue = new LinkedBlockingQueue<>();
        this.updateTaskQueue = new LinkedBlockingQueue<>();
        this.customizer = customizer;
        this.incrementalDataLoader = incrementalDataLoader;
        this.processUpdateTaskTimeoutMs = TimeUnit.SECONDS.toMillis(timeoutSec);
    }

    LocalCacheCore(String name, String scheduledFullUpdateCron, long timeoutSec, ILocalCacheCustomizer<T> customizer, ILocalCacheCustomizerType3<T> incrementalDataLoader, RedisConnectionFactory redisConnectionFactory, int checkpointUpdateIntervalSec) {
        this(name, scheduledFullUpdateCron, timeoutSec, customizer, incrementalDataLoader);
        if (checkpointUpdateIntervalSec < 1 || checkpointUpdateIntervalSec > 30 || 60 % checkpointUpdateIntervalSec != 0) {
            throw new IllegalArgumentException(LocalCacheType1.class.getSimpleName() + "(name=" + name + ") configure checkpointUpdateTask failed, checkpointUpdateIntervalSec ranges in 1/2/3/4/5/6/10/12/15/20/30");
        }
        if (null == redisConnectionFactory) {
            throw new IllegalArgumentException("redisConnectionFactory CANNOT be null");
        }
        this.redisConnectionFactory = redisConnectionFactory;
        this.checkpointUpdateCron = " */" + checkpointUpdateIntervalSec + " * * * * *";
    }

    /**
     * 检查参数并创建缓存，创建成功后系统会自动执行一次全量加载初始化，本方法只能调用一次
     */
    void start() {
        synchronized (LocalCacheCore.class) {
            if (CACHE_NAMES.contains(name) || CACHE_NAMES.contains(name + CHECKPOINT_UPDATE_KEY_SUFFIX) || CACHE_NAMES.contains(name + CHECKPOINT_UPDATE_TOPIC_SUFFIX)) {
                throw new IllegalArgumentException("CacheName [" + name + "] exists, please change another one");
            }
            CACHE_NAMES.add(name);

            processUpdateTask(true);

            log.info("[{}] register [scheduledFullUpdateTask] cron={}", name, scheduledFullUpdateCron);
            CronExecutorUtils.schedule(dispatcher, executor, scheduledFullUpdateCron, () -> {
                try {
                    updateTaskQueue.put(LocalCacheTaskType.SCHEDULED_FULL_UPDATE);
                } catch (Throwable ex) {
                    try {
                        customizer.onError(name, ex);
                    } catch (Throwable e) {
                        log.error("[{}] cronJob-exception", name, e);
                    }
                }
            });

            if (null != checkpointUpdateCron) {
                log.info("[{}] register [checkpointUpdateTask] cron={}", name, checkpointUpdateCron);
                CronExecutorUtils.schedule(dispatcher, executor, checkpointUpdateCron, () -> {
                    try {
                        updateTaskQueue.put(LocalCacheTaskType.CHECKPOINT_UPDATE);
                    } catch (Throwable ex) {
                        try {
                            customizer.onError(name, ex);
                        } catch (Throwable e) {
                            log.error("[{}] cronJob-exception", name, e);
                        }
                    }
                });
            }

            log.info("[{}] register [queueConsumerTask] cron={}", name, PROCESS_UPDATE_TASK_CRON);
            CronExecutorUtils.scheduleWithDelay(dispatcher, executor, PROCESS_UPDATE_TASK_CRON, PROCESS_UPDATE_TASK_DELAY_MS, () -> {
                processUpdateTask(false);
            });

            if (null != redisConnectionFactory) {
                RedisConnection conn = redisConnectionFactory.getConnection();
                log.info("[{}] register [redisPubSubTask] topic={}", name, redisTopic);
                conn.subscribe((message, pattern) -> {
                    executor.execute(() -> {
                        try {
                            String msg = new String(message.getBody());
                            log.info("[{}] redisPubSubTask1-received topic={} msg={}", name, redisTopic, msg);
                            checkpointMessageQueue.put(msg);
                        } catch (Throwable ex) {
                            log.error("[{}] redisPubSubTask2-exception topic={}", name, redisTopic, ex);
                            try {
                                customizer.onError(name, ex);
                            } catch (Throwable e) {
                                log.error("[{}] redisPubSubTask3-exception- topic={}", name, redisTopic, e);
                            }
                        }
                    });
                }, redisTopic.getBytes(StandardCharsets.UTF_8));
            }
        }
    }

    void processUpdateTask(boolean firstTime) {
        if (running) {
            log.debug("[{}] checkUpdate1-waitingQueueSize={}", name, updateTaskQueue.size());
            long now = System.currentTimeMillis();
            if (now >= nextTimeoutReportTimeMs) { // 执行时间超过 timeoutMs * n 时触发告警
                nextTimeoutReportTimeMs = now + processUpdateTaskTimeoutMs;
                try {
                    log.error("[{}] taskTimeout startAt={} timeoutMs={}ms", name, DatetimeUtils.format(processUpdateTaskStartTimeMs), processUpdateTaskTimeoutMs);
                    executor.execute(() -> customizer.onTimeout(name, processUpdateTaskStartTimeMs));
                } catch (Throwable ex) {
                    log.error("[{}] taskTimeout-exception", name, ex);
                }
            }
        } else {
            running = true;
            processUpdateTaskStartTimeMs = System.currentTimeMillis();
            nextTimeoutReportTimeMs = processUpdateTaskStartTimeMs + processUpdateTaskTimeoutMs;
            try {
                do {
                    // 如果是初始化时的首次加载，就默认当有一次全量加载任务要执行
                    LocalCacheTaskType tp = firstTime ? LocalCacheTaskType.SCHEDULED_FULL_UPDATE : updateTaskQueue.poll();
                    while (!updateTaskQueue.isEmpty()) {
                        // 把堆积的任务做合并处理，多个检查点更新可以合并成一个，一旦有一个【定时全量更新任务】，就可以合并成全量更新
                        LocalCacheTaskType p = updateTaskQueue.poll();
                        if (LocalCacheTaskType.SCHEDULED_FULL_UPDATE.equals(p)) {
                            tp = p;
                        }
                    }
                    T fromSource;
                    if (null == tp) {
                        log.debug("[{}] checkUpdate2.1-noQueueTask", name);
                        break;
                    } else {
                        String checkpointUpdateTagInRedis = "";
                        if (null != redisConnectionFactory) {
                            try (RedisConnection conn = redisConnectionFactory.getConnection()) {
                                byte[] bytes = conn.get(redisKey.getBytes(StandardCharsets.UTF_8));
                                if (null != bytes && bytes.length > 0) {
                                    checkpointUpdateTagInRedis = new String(bytes, StandardCharsets.UTF_8);
                                }
                            }
                        }
                        if (LocalCacheTaskType.SCHEDULED_FULL_UPDATE.equals(tp)) {
                            log.info("[{}] checkUpdate2.2-scheduledFullUpdateTask-START type={} cacheUpdateTag={} checkpointUpdateTagInRedis={}", name, tp.name(), checkpointUpdateTag, checkpointUpdateTagInRedis);
                            long startMs = System.currentTimeMillis();
                            try {
                                checkpointMessageQueue.clear();
                                fromSource = customizer.load();
                                if (null == fromSource) {
                                    throw new NullPointerException(ILocalCacheCustomizer.class.getSimpleName() + ".load() returns null is not allowed");
                                }
                                cache = fromSource;
                                checkpointUpdateTag = checkpointUpdateTagInRedis;
                            } finally {
                                log.info("[{}] checkUpdate2.2-scheduledFullUpdateTask-END cost={}ms type={} cacheUpdateTag={} checkpointUpdateTagInRedis={}", name, System.currentTimeMillis() - startMs, tp.name(), checkpointUpdateTag, checkpointUpdateTagInRedis);
                            }
                        } else if (LocalCacheTaskType.CHECKPOINT_UPDATE.equals(tp)) {
                            log.debug("[{}] checkUpdate2.3-checkpointUpdateTask type={} cacheUpdateTag={} checkpointUpdateTagInRedis={}", name, tp.name(), checkpointUpdateTag, checkpointUpdateTagInRedis);
                            if (!checkpointUpdateTagInRedis.equals(checkpointUpdateTag)) {
                                List<String> keys = new ArrayList<>();
                                while (!checkpointMessageQueue.isEmpty()) {
                                    String p = checkpointMessageQueue.poll();
                                    if (null != p) {
                                        keys.add(p);
                                    }
                                }
                                // 遇到全量更新标记，此时就不需要增量更新了，直接执行全量更新就行了
                                if (keys.contains(FULL_UPDATE_KEY)) {
                                    keys.clear();
                                }
                                if (keys.isEmpty()) {
                                    log.info("[{}] checkUpdate2.3.1-checkpointUpdateTask type={} -> fullUpdate-START cacheUpdateTag={} checkpointUpdateTagInRedis={}", name, tp.name(), checkpointUpdateTag, checkpointUpdateTagInRedis);
                                    long startMs = System.currentTimeMillis();
                                    try {
                                        fromSource = customizer.load();
                                        if (null == fromSource) {
                                            throw new NullPointerException(ILocalCacheCustomizer.class.getSimpleName() + " returns null is not allowed");
                                        }
                                        cache = fromSource;
                                    } finally {
                                        log.info("[{}] checkUpdate2.3.1-checkpointUpdateTask type={} -> fullUpdate-END cost={}ms", name, tp.name(), System.currentTimeMillis() - startMs);
                                    }
                                } else {
                                    log.info("[{}] checkUpdate2.3.2-checkpointUpdateTask type={} -> incrementalUpdateTask-START keys={} cacheUpdateTag={} checkpointUpdateTagInRedis={}", name, tp.name(), keys, checkpointUpdateTag, checkpointUpdateTagInRedis);
                                    long startMs = System.currentTimeMillis();
                                    try {
                                        if (null != incrementalDataLoader) {
                                            incrementalDataLoader.onIncrementalKeysReceived(cache, keys);
                                        }
                                    } finally {
                                        log.info("[{}] checkUpdate2.3.2-checkpointUpdateTask type={} -> incrementalUpdateTask-END cost={}ms keys={}", name, tp.name(), System.currentTimeMillis() - startMs, keys);
                                    }
                                }
                                checkpointUpdateTag = checkpointUpdateTagInRedis;
                            }
                        } else {
                            throw new IllegalStateException("It won't happen");
                        }
                    }
                } while (!updateTaskQueue.isEmpty());
            } catch (Throwable ex) {
                log.error("[{}] checkUpdate3-exception", name, ex);
                if (firstTime) {
                    throw new IllegalStateException(ex);
                }
                executor.execute(() -> customizer.onError(name, ex));
            } finally {
                running = false;
            }
        }
    }

    void markFullUpdate() {
        String tag = String.valueOf(UUID.randomUUID());
        try (RedisConnection conn = redisConnectionFactory.getConnection()) {
            conn.multi();
            conn.publish(redisTopic.getBytes(StandardCharsets.UTF_8), FULL_UPDATE_KEY.getBytes(StandardCharsets.UTF_8));
            conn.set(redisKey.getBytes(StandardCharsets.UTF_8), tag.getBytes(StandardCharsets.UTF_8), Expiration.seconds(REDIS_KEY_EXPIRE_SEC), RedisStringCommands.SetOption.upsert());
            conn.exec();
        }
        log.info("[{}] mark1-fullUpdateTask set->cacheUpdateTagInRedis={}", name, tag);
    }

    void markIncrementalUpdate(Collection<String> keys) {
        for (String key : keys) {
            if (null == key || key.equals(FULL_UPDATE_KEY)) {
                throw new IllegalArgumentException("key CANNOT be " + key);
            }
        }
        String tag = String.valueOf(UUID.randomUUID());
        try (RedisConnection conn = redisConnectionFactory.getConnection()) {
            conn.multi();
            for (String key : keys) {
                conn.publish(redisTopic.getBytes(StandardCharsets.UTF_8), key.getBytes(StandardCharsets.UTF_8));
            }
            conn.set(redisKey.getBytes(StandardCharsets.UTF_8), tag.getBytes(StandardCharsets.UTF_8), Expiration.seconds(REDIS_KEY_EXPIRE_SEC), RedisStringCommands.SetOption.upsert());
            conn.exec();
        }
        log.info("[{}] mark2-incrementalUpdateTask set->cacheUpdateTagInRedis={}, keys={}", name, tag, keys);
    }

    enum LocalCacheTaskType {

        /**
         * 定时全量更新任务，不管原数据有没有变化，都一定会执行
         */
        SCHEDULED_FULL_UPDATE,

        /**
         * 检查点更新任务，检查有没有对应的标记点变化，有才执行对应的全量/增量更新任务
         */
        CHECKPOINT_UPDATE
    }
}
